package com.yanfei.zero.application;


import java.util.HashMap;
import java.util.Map;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.connector.elasticsearch.sink.Elasticsearch7SinkBuilder;
import org.apache.flink.connector.elasticsearch.sink.ElasticsearchSink;
import org.apache.flink.elasticsearch7.shaded.org.apache.http.HttpHost;
import org.apache.flink.elasticsearch7.shaded.org.elasticsearch.action.index.IndexRequest;
import org.apache.flink.elasticsearch7.shaded.org.elasticsearch.client.Requests;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;


public class FlinkLearn {
        public static void main(String[] args) throws Exception {

        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
        .hostname("localhost")
        .port(3306)
        .databaseList("test")
        .username("root")
        .password("123456")
        .tableList("test")
        .deserializer(new JsonDebeziumDeserializationSchema())
        // .startupOptions(StartupOptions.initial())
        .includeSchemaChanges(true)
        .build();

        Configuration configuration = new Configuration();
        configuration.setInteger(RestOptions.PORT, 8881);
        StreamExecutionEnvironment  env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
        // enable checkpoint
        env.enableCheckpointing(30000);

        DataStream<String> source=  env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source")
        .returns(mySqlSource.getProducedType());
       
        
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        Table table =tableEnv.fromDataStream(source);
        
        // source.sinkTo(createEsSink());

        
      // 写到es中
 
 
        source.addSink(new CustomSink()).name("testsink");

        env.execute();

    }

private static IndexRequest createIndexRequest(String element) {
        System.out.println("输出数据"+element);
        Map<String, Object> json = new HashMap<>();
        json.put("data", element);

        return Requests.indexRequest()
        .index("my-index")
        .id(element)
        .source(json);
}

   public static ElasticsearchSink<String> createEsSink(){

       return new Elasticsearch7SinkBuilder<String>()
        .setHosts(new HttpHost("9200.gr3fada3.default.192.168.10.12.nip.io"))
        .setEmitter((element, context, indexer) ->
        indexer.add(createIndexRequest(element)))
        .build();
}

}